package com.javacv.plus.core.event;

import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
import java.util.concurrent.ForkJoinPool;

/**
 * 处理器事件总线
 * 支持事件的发布和订阅，实现组件间的解耦通信
 */
@Slf4j
public class ProcessorEventBus {
    
    private final ConcurrentHashMap<Class<?>, List<EventSubscriber<?>>> subscribers = new ConcurrentHashMap<>();
    private final Executor executor;
    private final boolean async;
    
    /**
     * 创建同步事件总线
     */
    public ProcessorEventBus() {
        this(false, null);
    }
    
    /**
     * 创建异步事件总线
     */
    public ProcessorEventBus(Executor executor) {
        this(true, executor);
    }
    
    private ProcessorEventBus(boolean async, Executor executor) {
        this.async = async;
        this.executor = executor != null ? executor : ForkJoinPool.commonPool();
    }
    
    /**
     * 订阅事件
     */
    public <T extends ProcessorEvent> void subscribe(Class<T> eventType, EventListener<T> listener) {
        subscribers.computeIfAbsent(eventType, k -> new CopyOnWriteArrayList<>())
                   .add(new EventSubscriber<>(listener));
        
        log.debug("事件订阅: {} -> {}", eventType.getSimpleName(), listener.getClass().getSimpleName());
    }
    
    /**
     * 取消订阅
     */
    public <T extends ProcessorEvent> void unsubscribe(Class<T> eventType, EventListener<T> listener) {
        List<EventSubscriber<?>> eventSubscribers = subscribers.get(eventType);
        if (eventSubscribers != null) {
            eventSubscribers.removeIf(subscriber -> subscriber.getListener().equals(listener));
            
            if (eventSubscribers.isEmpty()) {
                subscribers.remove(eventType);
            }
        }
        
        log.debug("取消事件订阅: {} -> {}", eventType.getSimpleName(), listener.getClass().getSimpleName());
    }
    
    /**
     * 发布事件
     */
    @SuppressWarnings("unchecked")
    public <T extends ProcessorEvent> void publish(T event) {
        List<EventSubscriber<?>> eventSubscribers = subscribers.get(event.getClass());
        
        if (eventSubscribers == null || eventSubscribers.isEmpty()) {
            log.debug("没有订阅者接收事件: {}", event.getClass().getSimpleName());
            return;
        }
        
        log.debug("发布事件: {} 给 {} 个订阅者", event.getClass().getSimpleName(), eventSubscribers.size());
        
        for (EventSubscriber<?> subscriber : eventSubscribers) {
            EventSubscriber<T> typedSubscriber = (EventSubscriber<T>) subscriber;
            
            if (async) {
                executor.execute(() -> deliverEvent(event, typedSubscriber));
            } else {
                deliverEvent(event, typedSubscriber);
            }
        }
    }
    
    /**
     * 投递事件给订阅者
     */
    private <T extends ProcessorEvent> void deliverEvent(T event, EventSubscriber<T> subscriber) {
        try {
            subscriber.getListener().onEvent(event);
        } catch (Exception e) {
            log.error("事件处理失败: {} by {}", 
                     event.getClass().getSimpleName(), 
                     subscriber.getListener().getClass().getSimpleName(), e);
        }
    }
    
    /**
     * 获取订阅者数量
     */
    public int getSubscriberCount(Class<?> eventType) {
        List<EventSubscriber<?>> eventSubscribers = subscribers.get(eventType);
        return eventSubscribers != null ? eventSubscribers.size() : 0;
    }
    
    /**
     * 清除所有订阅者
     */
    public void clear() {
        subscribers.clear();
        log.info("事件总线已清空");
    }
    
    /**
     * 事件订阅者包装器
     */
    private static class EventSubscriber<T extends ProcessorEvent> {
        private final EventListener<T> listener;
        
        public EventSubscriber(EventListener<T> listener) {
            this.listener = listener;
        }
        
        public EventListener<T> getListener() {
            return listener;
        }
    }
} 